package ws.client;

import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author patrick
 * @date 2021/11/20 17
 */
@Slf4j
public class WebSocketClient {
    private static final EventLoopGroup workerGroup = new NioEventLoopGroup();

    private Channel channel;

    private final boolean enableCompress;
    private final URL url;
    private final MessageInHandler messageInHandler;

    private final CountDownLatch finish;

    public static WebSocketClient create(String url, MessageInHandler messageInHandler) throws Exception {
        return create(true, url, messageInHandler);
    }

    public static WebSocketClient create(boolean enableCompress, String url, MessageInHandler messageInHandler) throws Exception {
        return new WebSocketClient(enableCompress, url, messageInHandler);
    }

    private WebSocketClient(boolean enableCompress, String urlStr, MessageInHandler messageInHandler) throws Exception {
        this.enableCompress = enableCompress;
        this.url = new URL(urlStr);
        this.messageInHandler = messageInHandler;
        this.finish = new CountDownLatch(1);

        init(this);

        this.finish.await(5, TimeUnit.SECONDS);
    }

    public URL url() {
        return url;
    }

    public MessageInHandler messageInHandler() {
        return messageInHandler;
    }

    public void send(String msg) {
        channel.writeAndFlush(new TextWebSocketFrame(msg));
    }

    public void close() {
        channel.close();
    }

    void finish() {
        finish.countDown();
        System.out.println("连接完成");
    }


    private void init(WebSocketClient webSocketClient) throws Exception {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup).channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(2 * 1024 * 1024, 4 * 1024 * 1024));
        bootstrap.handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("http-codec", new HttpClientCodec());
                if (enableCompress) {
                    pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
                }
                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
                pipeline.addLast(new WebSocketClientHandler(webSocketClient));
            }});

        ChannelFuture future = bootstrap.connect(new InetSocketAddress(url.getHost(), url.getPort())).sync();
        channel = future.channel();
    }

    public static void main(String[] args) throws Exception {

        WebSocketClient webSocketClient = WebSocketClient.create("http://10.121.18.255:8765/trade/option/usdc/public/v1", msg -> {
            System.out.println(msg);
        });

        Map<String, Object> ping = new HashMap<>();
        ping.put("id", String.valueOf(System.currentTimeMillis()));
        ping.put("op", "ping");
        webSocketClient.send(JSON.toJSONString(ping));
    }

}
